package cn.zwx.remoting.netty;

import cn.zwx.enumerate.HandleEnum;
import cn.zwx.exception.RemotingSendRequestException;
import cn.zwx.exception.RemotingTimeoutException;
import cn.zwx.remoting.RemotingCommand;
import cn.zwx.remoting.ResponseFuture;
import com.alibaba.fastjson.JSON;
import io.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 保存主要的成员变量为key和远程通道逻辑封装
 * @author 105286
 */
public class AbstractNettyRemoting {

    private final Logger logger = LoggerFactory.getLogger(AbstractNettyRemoting.class);
    /**
     *  存放唯一标示(UUID)和结果
     **/
    final ConcurrentHashMap<String, ResponseFuture> responseTable = new ConcurrentHashMap<>(8);

     void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand cmd) {
        if (cmd.getType() == HandleEnum.REQUEST_COMMAND.getCode()) {
            processRequestCommand(ctx, cmd);
        } else if (cmd.getType() == HandleEnum.RESPONSE_COMMAND.getCode()) {
            processResponseCommand(ctx, cmd);
        } else if (cmd.getType() == HandleEnum.PING_COMMAND.getCode()) {
            if (logger.isDebugEnabled()){
                logger.debug("receive ping from remoting client");
            }
        }
    }

    /**
     * 这个方法用于服务端实现
     * @param ctx ChannelHandlerContext
     * @param cmd RemotingCommand
     */
    protected void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {

    }

    private void processResponseCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        String xid = cmd.getXid();
        ResponseFuture responseFuture = responseTable.get(xid);
        responseFuture.putResponse(cmd);
    }

    /**
     * 同步调用
     * @param channel 连接方式
     * @param request RemotingCommand
     * @param timeoutMillis 超时时间
     * @return cn.zwx.remoting.RemotingCommand
     * @author zhangwenxue
     * @createTime 2022/6/20 11:10
     * @throws InterruptedException InterruptedException
     * @throws RemotingSendRequestException 发送远程调用异常
     * @throws RemotingTimeoutException 超时异常
     **/
    RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis) throws InterruptedException, RemotingSendRequestException {
        String xid = request.getXid();
        try {
            final SocketAddress addr = channel.remoteAddress();
            ResponseFuture responseFuture = new ResponseFuture(xid, timeoutMillis, null);
            responseTable.put(xid, responseFuture);
            channel.writeAndFlush(request).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    responseFuture.setSendRequestOk(Boolean.TRUE);
                    return;
                }
                responseFuture.setSendRequestOk(Boolean.FALSE);
                responseFuture.setCause(future.cause());
                responseFuture.putResponse(null);
            });
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) {
                if (!responseFuture.isSendRequestOk()) {
                    throw new RemotingSendRequestException(addr.toString(), responseFuture.getCause());
                }
            }
            return responseCommand;
        } finally {
            responseTable.remove(xid);
        }
    }

    /**
     * 扫描结果列表，将超时的结果释放
     */
    protected void scanResponseTable() {
        final List<ResponseFuture> rfList = new LinkedList<>();
        Iterator<Map.Entry<String, ResponseFuture>> iterator = this.responseTable.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ResponseFuture> next = iterator.next();
            ResponseFuture rep = next.getValue();
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                rep.release();
                iterator.remove();
                rfList.add(rep);
            }
        }
        for (ResponseFuture rf : rfList) {
            executeInvokeCallback(rf);
        }
    }

    /**
     *  处理回调方法
     * @param rf 返回参数
     * @return void
     * @author zhangwenxue
     * @createTime 2022/6/20 11:19
     **/
    private void executeInvokeCallback(ResponseFuture rf) {
        rf.executeInvokeCallback();
    }

    /**
     *  返回参数处理统一方法
     * @author zhangwenxue
     * @createTime 2022/6/20 11:18
     **/
    protected static class RemotingCommandHandle extends ChannelOutboundHandlerAdapter {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
            ctx.writeAndFlush(JSON.toJSONString(msg));
        }
    }

}
